package cn.tank.data.sync.redis;

import cn.hutool.core.lang.Assert;
import cn.tank.data.sync.core.MessageAcceptService;
import cn.tank.data.sync.core.common.Message;
import cn.tank.data.sync.core.common.Subscriber;
import com.alibaba.fastjson.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/************************************************
 * @Description:redis消息订阅器
 * @Version: 1.0
 ************************************************/
public class RedisMessageSubscriber implements MessageAcceptService {
    private final static Logger log = LoggerFactory.getLogger(RedisMessageSubscriber.class);

    private final RedisTemplate redisTemplate;

    private final RedisMessageListenerContainer container;

    private final Map<String, MessageListener> listenerMap = new ConcurrentHashMap<>();


    public RedisMessageSubscriber(RedisTemplate redisTemplate) {
        Assert.notNull(redisTemplate);

        this.redisTemplate = redisTemplate;

        this.container = createContainer(redisTemplate.getConnectionFactory());
    }

    public void createSubscriber(String channel, Subscriber subscriber) {
        if (subscriber == null) {
            throw new RuntimeException("订阅者对象不能为空");
        }

        listenerMap.putIfAbsent(subscriber.getId(), (message, pattern) -> {
            String body = new String(message.getBody());
            receiveMessage(channel, subscriber, JSONObject.toJavaObject(JSONObject.parseObject(body), Message.class));
        });

        container.addMessageListener(listenerMap.get(subscriber.getId()), new PatternTopic(channel));

        if (log.isInfoEnabled()) {
            log.info("订阅者{}成功订阅频道{}", subscriber.getName(), channel);
        }

        if (!container.isRunning()) {
            container.start();
        }
    }

    public void deleteSubscriber(String channel, Subscriber subscriber) {

        container.removeMessageListener(listenerMap.get(subscriber.getId()));

        if (log.isInfoEnabled()) {
            log.info("订阅者{}取消订阅频道{}", subscriber.getName(), channel);
        }
    }

    /**
     * 创建监听容器
     *
     * @param redisConnectionFactory
     * @return
     */
    private RedisMessageListenerContainer createContainer(RedisConnectionFactory redisConnectionFactory) {

        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(redisConnectionFactory);
        container.setErrorHandler(Throwable::printStackTrace);
        container.afterPropertiesSet();

        return container;
    }
}
